<?php
/**
 * CrawlerTask.php class file.
 * @author wudg <544038230@qq.com>
 * @date 2020/10/24- 5:02 下午
 */
namespace console\components\tasks;


use common\components\misc\ImportLogger;
use console\components\helpers\ConsoleHelper;
use yii\base\InvalidArgumentException;
use yii\base\NotSupportedException;
use yii\di\Instance;
use yii\helpers\FileHelper;
use yii\helpers\VarDumper;
use yii\queue\InvalidJobException;
use yii\queue\JobInterface;
use yii\queue\cli\Queue;
use yii\queue\serializers\PhpSerializer;
use yii\queue\serializers\SerializerInterface;
use yii\redis\Connection;
use Yii;
/**
 * 多进程执行yii-queue 队列任务封装
 * Class CrawlerTask
 * @package console\components\tasks
 */
class CrawlerTask extends Queue
{
    /**
     * @var Connection|array|string
     */
    public $redis = 'redis';
    /**
     * 队列名称
     * @var string
     */
    public $channel = 'queueSpiderTb';

    /**
     * @var SerializerInterface|array
     */
    public $serializer = PhpSerializer::class;

    public $logger;

    /**
     * 执行真正的采集页面任务
     * @param array $args
     * @return bool
     * @throws \yii\base\InvalidConfigException
     * @throws \yii\base\Exception
     */
    public function run()
    {
        $this->logger = new ImportLogger($this->_createFilePath());
        $this->redis = Instance::ensure($this->redis, Connection::class);

        $this->serializer = Instance::ensure($this->serializer, SerializerInterface::class);

        if (($payload = $this->reserve(0)) !== null) {
            list($id, $message, $ttr, $attempt) = $payload;
            $this->delete($id);
            if ($this->handleMessage($id, $message, $ttr, $attempt)) {
                $this->logger->add('采集任务['.$id.']执行完成');
                Yii::$app->db->close();
            }
        }
    }

    /**
     * 采集log目录初始化
     * @return bool|string
     * @throws \yii\base\Exception
     */
    protected function _createFilePath()
    {
        $path = Yii::getAlias('@runtime/task/spider/');

        //目录如果不存就先创建目录
        if(!is_dir($path)) {
            FileHelper::createDirectory($path);
        }

        return $path;
    }

    /**
     * @param int $timeout timeout
     * @return array|null payload
     */
    protected function reserve($timeout)
    {
        // Moves delayed and reserved jobs into waiting list with lock for one second
        if ($this->redis->set("$this->channel.moving_lock", true, 'NX', 'EX', 1)) {
            $this->moveExpired("$this->channel.delayed");
            $this->moveExpired("$this->channel.reserved");
        }

        // Find a new waiting message
        $id = null;
        if (!$timeout) {
            $id = $this->redis->rpop("$this->channel.waiting");
        } elseif ($result = $this->redis->brpop("$this->channel.waiting", $timeout)) {
            $id = $result[1];
        }
        if (!$id) {
            return null;
        }

        $payload = $this->redis->hget("$this->channel.messages", $id);
        list($ttr, $message) = explode(';', $payload, 2);
        $this->redis->zadd("$this->channel.reserved", time() + (int)$ttr, $id);
        $attempt = $this->redis->hincrby("$this->channel.attempts", $id, 1);

        return [$id, $message, $ttr, $attempt];
    }

    /**
     * @param string $from
     */
    protected function moveExpired($from)
    {
        $now = time();
        if ($expired = $this->redis->zrevrangebyscore($from, $now, '-inf')) {
            $this->redis->zremrangebyscore($from, '-inf', $now);
            foreach ($expired as $id) {
                $this->redis->rpush("$this->channel.waiting", $id);
            }
        }
    }

    /**
     * Deletes message by ID.
     *
     * @param int $id of a message
     */
    protected function delete($id)
    {
        $this->redis->zrem("$this->channel.reserved", $id);
        $this->redis->hdel("$this->channel.attempts", $id);
        $this->redis->hdel("$this->channel.messages", $id);
    }


    /**
     * @inheritdoc
     */
    public function status($id)
    {
        if (!is_numeric($id) || $id <= 0) {
            throw new InvalidArgumentException("Unknown message ID: $id.");
        }

        if ($this->redis->hexists("$this->channel.attempts", $id)) {
            return 2;
        }

        if ($this->redis->hexists("$this->channel.messages", $id)) {
            return 1;
        }

        return 3;
    }

    /**
     * Clears the queue.
     *
     * @since 2.0.1
     */
    public function clear()
    {
        while (!$this->redis->set("$this->channel.moving_lock", true, 'NX')) {
            usleep(10000);
        }
        $this->redis->executeCommand('DEL', $this->redis->keys("$this->channel.*"));
    }

    /**
     * Unserializes.
     *
     * @param string $id of the job
     * @param string $serialized message
     * @return array pair of a job and error that
     */
    public function unserializeMessage($serialized)
    {
        try {
            $job = $this->serializer->unserialize($serialized);
        } catch (\Exception $e) {
            return [null, new InvalidJobException($serialized, $e->getMessage(), 0, $e)];
        }

        if ($job instanceof JobInterface) {
            return [$job, null];
        }

        return [null, new InvalidJobException($serialized, sprintf(
            'Job must be a JobInterface instance instead of %s.',
            VarDumper::dumpAsString($job)
        ))];
    }

    /**
     * Removes a job by ID.
     *
     * @param int $id of a job
     * @return bool
     * @since 2.0.1
     */
    public function remove($id)
    {
        while (!$this->redis->set("$this->channel.moving_lock", true, 'NX', 'EX', 1)) {
            usleep(10000);
        }
        if ($this->redis->hdel("$this->channel.messages", $id)) {
            $this->redis->zrem("$this->channel.delayed", $id);
            $this->redis->zrem("$this->channel.reserved", $id);
            $this->redis->lrem("$this->channel.waiting", 0, $id);
            $this->redis->hdel("$this->channel.attempts", $id);

            return true;
        }

        return false;
    }

    /**
     * @inheritDoc
     */
    protected function pushMessage($message, $ttr, $delay, $priority)
    {
        if ($priority !== null) {
            throw new NotSupportedException('Job priority is not supported in the driver.');
        }

        $id = $this->redis->incr("$this->channel.message_id");
        $this->redis->hset("$this->channel.messages", $id, "$ttr;$message");
        if (!$delay) {
            $this->redis->lpush("$this->channel.waiting", $id);
        } else {
            $this->redis->zadd("$this->channel.delayed", time() + $delay, $id);
        }

        return $id;
    }
}
